package cn.com.guage.flink.transformation;

import java.util.ArrayList;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class KeyByTransformation {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        ArrayList<Tuple3<String, String, Integer>> data = new ArrayList<Tuple3<String, String, Integer>>();
        data.add(new Tuple3<String, String, Integer>("男", "老王", 80));
        data.add(new Tuple3<String, String, Integer>("男", "小王", 25));
        data.add(new Tuple3<String, String, Integer>("男", "老李", 85));
        data.add(new Tuple3<String, String, Integer>("男", "小李", 20));

        DataStreamSource<Tuple3<String,String,Integer>> peoples = env.fromCollection(data);
		
		//求年龄最小的人
        SingleOutputStreamOperator<Tuple3<String,String,Integer>> minResult = peoples.keyBy(0).min(2);
        minResult.print();

        env.execute("KeyByTransformation");
	}
}
